package com.dabai.iot.mqtt.event;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.dabai.common.core.utils.Func;
import com.dabai.common.core.utils.StringUtils;
import com.dabai.iot.common.constant.IotKeyConstant;
import com.dabai.iot.common.enums.DataType;
import com.dabai.iot.common.enums.TopicSourceType;
import com.dabai.iot.common.model.DeviceInfo;
import com.dabai.iot.common.model.IotMqttMessage;
import com.dabai.iot.core.api.RemoteDeviceService;
import com.dabai.iot.core.api.RemoteMessageHandlerService;
import com.dabai.iot.core.api.RemoteProductService;
import com.dabai.iot.core.entity.DeviceEntity;
import com.dabai.iot.core.entity.ProductTopicEntity;
import com.dabai.iot.mqtt.model.DeviceConnectionSession;
import com.dabai.iot.mqtt.model.MqttMessageContext;
import com.dabai.iot.mqtt.server.processor.MessageToDeviceProcessor;
import com.dabai.iot.mqtt.topics.SysTopics;
import com.dabai.iot.mqtt.topics.TopicDirection;
import com.dabai.iot.mqtt.util.ClientUtil;
import com.dabai.iot.mqtt.util.MqttSessionManager;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.mqtt.MqttEndpoint;
import io.vertx.mqtt.messages.MqttPublishMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

/**
 * 服务端消息监听
 *
 * @author kev1n
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ServerMessageListener implements IMqttMessageListener {

    private final MessageToDeviceProcessor messageToDeviceProcessor;

    @DubboReference(stub = "true")
    private final RemoteProductService remoteProductService;
    @DubboReference(stub = "true")
    private final RemoteDeviceService remoteDeviceService;
    @DubboReference
    private final RemoteMessageHandlerService remoteMessageHandlerService;

    /**
     * 监听到服务端发布消息
     */
    @Override
    public void onMessage(MqttEndpoint endpoint, MqttPublishMessage publishMessage) {
        String traceId = IdWorker.get32UUID();
        String serverClientId = endpoint.clientIdentifier();
        String serverTopic = publishMessage.topicName();
        String message = String.valueOf(publishMessage.payload());

        log.info("MQTT Server ====> 收到【服务端】发布消息: traceId={} clientId={} topic={} payload: {}", traceId, serverClientId, serverTopic, message);
        if (StringUtils.isBlank(message)) {
            return;
        }

        JSONObject json = JSON.parseObject(message);
        String productKey = json.getString(IotKeyConstant.PRODUCT_KEY);
        String deviceSn = json.getString(IotKeyConstant.DEVICE_SN);
        long timestamp = Func.toLong(json.getLong(IotKeyConstant.REPORT_TIME), System.currentTimeMillis());
        Date now = new Date(timestamp);

        DeviceEntity device = remoteDeviceService.getDeviceCacheBySn(productKey, deviceSn);
        if (device == null) {
            return;
        }

        String tenantId = device.getTenantId();
        Long productId = device.getProductId();
        Long deviceId = device.getId();
        String dataType = device.getDataType();

        JSONObject contentJson = json.getJSONObject("content");
        String topic = contentJson.getString("topic");
        Integer qos = Func.toInt(contentJson.getInteger("qos"), 0);

        String payload = "";
        if (DataType.JSON.getValue().equals(device.getDataType())) {
            JSONObject payloadJson = contentJson.getJSONObject("payload");
            payload = payloadJson.toJSONString();
        } else if (DataType.Plain.getValue().equals(device.getDataType())) {
            payload = contentJson.getString("payload");
        }

        String propertySetTopic = SysTopics.PROPERTY_SET
            .replace("${productKey}", device.getProductKey())
            .replace("${deviceSn}", device.getDeviceSn());

        if (propertySetTopic.equals(topic)) {
            // 发布消息到设备
            publishMessage(traceId, productKey, deviceSn, topic, payload, qos);
        } else {
            // 查询设备订阅的 topic
            List<ProductTopicEntity> productTopics = remoteProductService.getTopicList(device.getProductId(), TopicSourceType.custom.getValue(), TopicDirection.SUB.getValue());
            if (Func.isNotEmpty(productTopics)) {
                List<String> collect = productTopics.stream()
                    .map(t -> {
                        String productTopic = t.getTopic();
                        System.out.println(productTopic);
                        String topic1 = productTopic.replaceAll(SysTopics.DEVICE_SN_ESCAPE, "%s");
                        System.out.println(topic1);
                        return String.format(topic1, deviceSn);
                    }).toList();

                collect.forEach(System.out::println);

                if (collect.contains(topic)) {
                    // 发布消息到设备
                    publishMessage(traceId, productKey, deviceSn, topic, payload, qos);
                }
            }
        }

        // 调用 core 模块服务端发布消息处理
        DeviceInfo deviceInfo = new DeviceInfo(tenantId, productId, productKey, deviceId, deviceSn, dataType);
        IotMqttMessage mqttMessage = new IotMqttMessage(traceId, topic, payload, qos, dataType, now);
        remoteMessageHandlerService.serverMessageHandler(deviceInfo, mqttMessage);
    }

    /**
     * 发布消息到设备
     */
    private void publishMessage(String traceId, String productKey, String deviceSn, String topic, String payload, Integer qos) {
        String clientId = ClientUtil.deviceClientId(productKey, deviceSn);
        DeviceConnectionSession session = MqttSessionManager.getInstance().getSession(clientId);
        if (session == null) {
            return;
        }

        MqttQoS mqttQoS = MqttQoS.AT_MOST_ONCE;
        if (!qos.equals(0)) {
            mqttQoS = MqttQoS.AT_LEAST_ONCE;
        }

        MqttMessageContext mqttMessageContext = new MqttMessageContext();
        mqttMessageContext.setMqttEndpoint(session.getMqttEndpoint());
        mqttMessageContext.setTraceId(traceId);
        mqttMessageContext.setTopic(topic);
        mqttMessageContext.setPayload(payload);
        mqttMessageContext.setQosLevel(mqttQoS);

        messageToDeviceProcessor.add(mqttMessageContext);
    }

}
